package org.apache.spark.akka

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import org.apache.spark.common.Constant
import org.apache.spark.common.Message.{CheckTimeOut, ClientInfo, Heartbeat, NotifyOffline, NotifyOnline, RegisterClient, RegisteredClient}

import scala.collection.mutable

/**
 * @author: chenzhidiao
 * @date: 2021/1/4 10:37
 * @description:
 * @version: 1.0
 */
class MyServer(var hostname: String, var port: Int) extends Actor {
  //服务端用于保存客户端注册信息的容器
  var clientsInfo = new mutable.HashMap[String, ClientInfo]()
  //服务端用于保存去重后的所有客户端信息
  var clients = new mutable.HashSet[ClientInfo]()
  //存储所有注册客户端的ActorRef代理
  var clientsActorRef = new mutable.HashMap[String,ActorSelection]()

  override def preStart(): Unit = {
    import context.dispatcher

    import scala.concurrent.duration._

    //启动一个定时任务， 每隔三秒检查一次是否有客户端心跳超时
    context.system.scheduler.schedule(0 millis, 3000 millis, self, CheckTimeOut)

  }

  override def receive: Receive = {

    case CheckTimeOut => {
      val currentTime = System.currentTimeMillis()
      clients.filter(client => {
        val heartBeatTimeout = 10000
        val bool = currentTime - client.lastHeartBeatTime > heartBeatTimeout
        //如果客户端心跳超时，通知其他客户端该客户端下线
        if (bool) {
          clients.foreach(onlineClient => {
            if (onlineClient.clientid != client.clientid) {
              //获取通知对象的ActorRef代理
              var clientRef: ActorSelection = clientsActorRef.get(onlineClient.clientid).get

              clientRef ! NotifyOffline(client.clientid)
            }
          })
          println(s"客户端 ${client.clientid} 下线")
        }
        bool
      }).foreach(deadclient => {
        clients -= deadclient
        clientsInfo.remove(deadclient.clientid)
      })
      println("当前在线的客户端数为：" + clientsInfo.size + "\t分别是：" + clients.map(x => "\r\n\t"+x.toString)
        .mkString(","));
    }

    case Heartbeat(clientid) => {
      val currentTime = System.currentTimeMillis()
      var newInfo = clientsInfo(clientid)
      newInfo.lastHeartBeatTime = currentTime
      clientsInfo(clientid) = newInfo
      clients += newInfo
    }

    case RegisterClient(clientid, hostname, port, actorSystemName, actorName) => {
      var client = new ClientInfo(clientid, hostname, port, actorSystemName, actorName)
      println(s"客户端 ${client.clientid} 上线")
      clientsInfo.put(clientid, client)
      clients += client
      val clientRef: ActorSelection = context.actorSelection(s"akka.tcp://${
        actorSystemName
      }@${hostname}:${port}/user/${actorName}")

      clientsActorRef.put(clientid,clientRef)

      //通知其它在线客户端有新的客户端上线
      clients.foreach(onlineClient => {
        if (onlineClient.clientid != clientid) {
          val clientRef = clientsActorRef.get(onlineClient.clientid).get

          clientRef ! NotifyOnline(clientid)
        }
      })
      //sender() 谁给我发消息，sender方法返回的就是谁
      sender() ! RegisteredClient(hostname + ":" + port)

    }
  }
}


object MyServer {
  def main(args: Array[String]): Unit = {
    val str =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = localhost
         |akka.remote.netty.tcp.port = 6789
      """.stripMargin
    val conf = ConfigFactory.parseString(str)

    //创建服务端的ActorSystem
    val actorSystem = ActorSystem(Constant.SAS, conf)

    //启动了一个actor ： MyServer
    actorSystem.actorOf(Props(new MyServer("localhost", 6789)), Constant.SA)

  }
}
